package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Code03Table {
    Connection connection;

    @Before
    public void getConnection() {
        Configuration configuration = new Configuration();

        // 注意：需要在当前平台中配置node1 node2的主机映射
        configuration.set("hbase.zookeeper.quorum", "192.168.163.100,node1,node2");
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void putData() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:stu"));

        // put '表','rowkey','列'，‘value’
        Put put = new Put("1500001".getBytes());

//        new Cell();
//        put.add()
        //  public Put addColumn(byte [] family, byte [] qualifier, byte [] value)
        put.addColumn(
                "info".getBytes(),
                "name".getBytes(),
                "张三".getBytes()
        );

        put.addColumn(
                "info".getBytes(),
                "age".getBytes(),
                "18".getBytes()
        );

        put.addColumn(
                "info".getBytes(),
                "gender".getBytes(),
                "女".getBytes()
        );

        put.addColumn(
                "info".getBytes(),
                "clazz".getBytes(),
                "文科一班".getBytes()
        );

        put.addColumn(
                "score".getBytes(),
                "语文".getBytes(),
                "90".getBytes()
        );

        table.put(put);
    }

    @Test
    public void putData2() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:stu"));

        // put '表','rowkey','列'，‘value’
        Put put = new Put("1500002".getBytes());

//        new Cell();
//        put.add()
//        public KeyValue(final byte [] row, final byte [] family,
//            final byte [] qualifier, final byte [] value)
        KeyValue keyValue = new KeyValue();
        put.add(
                new KeyValue(
                        "1500002".getBytes(),
                        "info".getBytes(),
                        "name".getBytes(),
                        "李四".getBytes()
                )
        );

        table.put(put);
    }

    @Test
    public void putData3() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:split_stu"));

        // create 'api:split_stu', 'info', 'score',{SPLITS_FILE => '/usr/local/soft/hbase-2.1.9/splits_file/stu_split_info.txt'}
        // 需要在stu_split_info.txt文件中添加预分区的规则

        // IO 流读取每一行数据
        FileReader fileReader = new FileReader("data/student/students.txt");

        BufferedReader bufferedReader = new BufferedReader(fileReader);
        List<Put> putList = new ArrayList<>();

        String line = "";
        while ((line = bufferedReader.readLine()) != null) {
            try {
                String[] split = line.split(",");
                Put put = new Put(
                        Bytes.toBytes(split[0])
                );
                put.add(new KeyValue(
                        Bytes.toBytes(split[0]),
                        Bytes.toBytes("info"),
                        Bytes.toBytes("name"),
                        Bytes.toBytes(split[1])
                ));
                put.add(new KeyValue(
                        Bytes.toBytes(split[0]),
                        Bytes.toBytes("info"),
                        Bytes.toBytes("age"),
                        Bytes.toBytes(split[2])
                ));
                put.add(new KeyValue(
                        Bytes.toBytes(split[0]),
                        Bytes.toBytes("info"),
                        Bytes.toBytes("gender"),
                        Bytes.toBytes(split[3])
                ));
                put.add(new KeyValue(
                        Bytes.toBytes(split[0]),
                        Bytes.toBytes("info"),
                        Bytes.toBytes("clazz"),
                        Bytes.toBytes(split[4])
                ));
                putList.add(
                        put
                );
            } catch (ArrayIndexOutOfBoundsException e) {
                System.out.println("问题行：" + line);
            }
        }
        table.put(putList);

        bufferedReader.close();
    }

    /*
    1500100001,施笑槐,22,女,文科六班
        1500100001,1000001,98
        1500100001,1000002,5
        1500100001,1000003,137
        1500100001,1000004,29
        1500100001,1000005,85
        1500100001,1000006,52
     */


    @Test
    public void putScoreData() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:stu"));

        FileReader fileReader = new FileReader("data/student/score.txt");
        BufferedReader bufferedReader = new BufferedReader(fileReader);

        ArrayList<Put> puts = new ArrayList<>();

        String oneScore = "";
        while ((oneScore = bufferedReader.readLine()) != null) {
            String[] split = oneScore.split(",");
            Put put = new Put(split[0].getBytes());
            put.addColumn("score".getBytes()
                    , split[1].getBytes()
                    , split[2].getBytes()
            );
            puts.add(put);
        }

        // 上上传一个List中的数据
        table.put(puts);
        // 如何实现每1000条数据，上传一次putList
        // 如何实现从Mysql中读取数据，写入至Hbase中(ETL)

    }


    @After
    public void close() {
        try {
            connection.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
